package com.google.code.echronos;

import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonGenerator;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;

final class EventSourceImpl implements EventSource {
	private static final String DEFAULT_DB = "events.db";

	private final ObjectMapper objectMapper;

	private final Path path;
	private final List<EventConsumer> eventConsumers = new ArrayList<>();

	private boolean started;
	private boolean stopped;

	private ObjectOutputStream outputStream;


	public EventSourceImpl(final Path path) throws DatabaseNotDirectoryException {
		if (!Files.isDirectory(path)) {
			throw new DatabaseNotDirectoryException(path);
		}
		this.path = path;

		final JsonFactory jsonFactory = new JsonFactory();
		jsonFactory.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET);
		jsonFactory.disable(JsonParser.Feature.AUTO_CLOSE_SOURCE);
		objectMapper = new ObjectMapper(jsonFactory);
	}

	@Override
	public void register(final EventConsumer eventConsumer) {
		if (started) {
			throw new AlreadyStartedException();
		}
		eventConsumers.add(eventConsumer);
	}

	@Override
	public void start() {
		start(System.currentTimeMillis());
	}

	@Override
	public void start(final long timestamp) {
		if (started) {
			throw new AlreadyStartedException();
		}

		final Path eventsDb = path.resolve(DEFAULT_DB);

		// replay all events
		if (Files.exists(eventsDb)) {
			try (final InputStream inputStream = Files.newInputStream(eventsDb)) {
				final ObjectInputStream objectStream = new ObjectInputStream(inputStream);
				//noinspection InfiniteLoopStatement
				for (;;) {
					final Event event = read(objectStream);
					if (event.getTimestamp() >= timestamp) {
						multiplex(event);
					}
				}
			} catch (EOFException e) {
				// no more events to read
			} catch (IOException e) {
				throw new IllegalStateException(e);
			}
		}

		// open output stream
		try {
			outputStream = new ObjectOutputStream(Files.newOutputStream(eventsDb, StandardOpenOption.APPEND, StandardOpenOption.CREATE));
		} catch (IOException e) {
			throw new IllegalStateException(e);
		}

		started = true;
	}

	private Event read(final ObjectInputStream stream) throws IOException {
		final Long timestamp = stream.readLong();
		final String identifier = stream.readUTF();
		final String payloadData = stream.readUTF();

		final Object payload;
		if (payloadData.isEmpty()) {
			payload = null;
		} else {
			payload = objectMapper.readValue(payloadData, Object.class);
		}
		return new EventImpl(identifier, timestamp, payload);
	}

	@Override
	public void trigger(final String identifier, final Object payload) {
		if (!started) {
			throw new NotYetStartedException();
		}
		if (stopped) {
			throw new AlreadyStoppedException();
		}
		final Event event = new EventImpl(identifier, System.currentTimeMillis(), payload);
		try {
			write(event);
		} catch (IOException e) {
			throw new IllegalStateException(e);
		}
		multiplex(event);
	}

	private void write(final Event event) throws IOException {
		outputStream.writeLong(event.getTimestamp());
		outputStream.writeUTF(event.getIdentifier());

		if (event.getPayload() == null) {
			outputStream.writeUTF("");
		} else {
			final StringWriter stringWriter = new StringWriter();
			objectMapper.writeValue(stringWriter, event.getPayload());
			outputStream.writeUTF(stringWriter.toString());
		}
	}

	private void multiplex(final Event event) {
		for (final EventConsumer eventConsumer : eventConsumers) {
			try {
				eventConsumer.consumeEvent(event);
			} catch (Exception ignored) {
				// we can not allow the flow to be disrupted
			}
		}
	}

	@Override
	public void stop() {
		if (!started) {
			throw new NotYetStartedException();
		}

		try {
			outputStream.close();
		} catch (IOException ignored) {
		}
		outputStream = null;

		stopped = true;
	}
}
